package com.lefu.risk.storm.generdata;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

public class ReducerBolt extends BaseRichBolt{

	private static final long serialVersionUID = 1L;
	
	private OutputCollector collector;
	
	private FileWriter out;
	
	private File file;

	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
	}

	@Override
	public void execute(Tuple input) {
		StringBuffer buffer = new StringBuffer();
		String[] arr = (String[])input.getValueByField("arr");
		String sort = input.getStringByField("sort");
		Long start = input.getLongByField("start");
		for (String string : arr) {
			buffer.append(string).append("\t");
		}
		
		try {
			String fileName = "F:\\"+sort+".txt";
			
			try {
				if(file == null){
					file = new File(fileName);
					if(file.exists()){
						file.createNewFile();
					}
				}
				if(out == null){
					out = new FileWriter(file, true);
				}
				out.write(buffer.toString()+"\r\n");
				out.flush();
			} catch (Exception e) {
				e.printStackTrace();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		System.out.println("ShuffGenerData -------> ReducerBolt cost ："+(System.currentTimeMillis()-start));
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		
	}
}
